/*
 * Hyuga: A Density-Grid Stream Clustering Platform.
 * Copyright (C) 2014 PUC-Rio/Laboratory for Advanced Collaboration
 *
 * Hyuga is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * Hyuga is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with Hyuga.  If not, see <http://www.gnu.org/licenses/>.
 */
package br.pucrio.inf.lac.konoha.hyuga.others.csv;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.io.PrintWriter;
import java.io.RandomAccessFile;
import java.nio.channels.FileChannel;
import java.nio.channels.FileLock;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.log4j.ConsoleAppender;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.apache.log4j.PatternLayout;

import br.pucrio.inf.lac.konoha.hyuga.common.DBSCAN;
import br.pucrio.inf.lac.konoha.hyuga.common.Pair;
import br.pucrio.inf.lac.konoha.hyuga.events.csv.CSVClusterEvent;
import br.pucrio.inf.lac.konoha.hyuga.events.csv.CSVLocationUpdateEvent;

/**
 * A DBSCAN CSV clustering routine. The csv file must have the following columns:
 * id,latitude,longitude
 * 
 * Sample File:
 * uuid, latitude,          longitude,         timestamp
 * 286,  -22.9559986550780, -43.1922377359216, 70462
 * 973,  -22.9603834270102, -43.2000845051359, 70462
 * 2742, -22.9728237658448, -43.1907249148011, 704637
 * 4075, -22.9788327380116, -43.1908675079309, 70464
 * 878,  -22.9811741072085, -43.2333375518404, 70464
 * 3431, -22.9703160507337, -43.1907072643957, 704650
 * 341,  -22.9656428061742, -43.2163599814267, 70466
 * 5028, -22.9580526401979, -43.2050608584764, 704664
 *
 * @author Marcos Roriz <mroriz@inf.puc-rio.br>
 */
public class CSVDBSCAN {

    /** LOGGER. */
	private static Logger LOGGER = Logger.getLogger("br.pucrio.inf.lac.konoha.hyuga");
	
	private double eps;
	private int    minPts;
	
	// Start / End Period
	private double startPeriod;
	private double endPeriod;
	
	// Input File
	private File inputFile;
	
	// Output File
	private File outputFile;
	
	public CSVDBSCAN(String inName, double eps, int minPts, double time, double period, String outName) throws FileNotFoundException {
	    // DBSCAN variables
	    this.eps    = eps;
	    this.minPts = minPts;
	    
	    // CSV variables
	    this.startPeriod = time - period;
	    this.endPeriod   = time;
        this.inputFile   = new File(inName);
        this.outputFile  = new File(outName);
        
        if (startPeriod < 0) startPeriod = 0;
        
        if (!inputFile.exists()) {
            throw new FileNotFoundException();
        }
    }
	
	public void cluster() throws IOException {
	    long start = System.currentTimeMillis();
	    Map<String, CSVLocationUpdateEvent> nodes = new HashMap<String, CSVLocationUpdateEvent>();
	    
	    FileReader fr = new FileReader(inputFile);
	    BufferedReader buffer = new BufferedReader(fr);
	    buffer.readLine(); // header
	
	    // Local variables (reused)
	    boolean finished = false;
	    double  minlat   = -90;
        double  maxlat   = +90;
        double  minlng   = -180;
        double  maxlng   = +180;
        minlat = -23.069387;
        maxlat = -22.766447;
        minlng = -43.784981;
        maxlng = -43.142967;
        
        String line = "";
        while ((line = buffer.readLine()) != null) {
            String[] data = line.split(",");
            
            if (Integer.parseInt(data[3]) >= startPeriod) {
                double longitude = Double.parseDouble(data[2]);
                double latitude = Double.parseDouble(data[1]);
                
                if (minlng < longitude && longitude < maxlng && minlat < latitude && latitude < maxlat) {
                    nodes.put(data[0], new CSVLocationUpdateEvent(data[0], latitude, longitude));
                }
            }
            
            if (Integer.parseInt(data[3]) >= endPeriod) {
                break;
            }
        }
	    buffer.close();

	    System.out.println("Number of Points: " + nodes.size());
        
	    DBSCAN dbscan = new DBSCAN(eps, minPts);
        List<CSVClusterEvent> clusters = dbscan.cluster(nodes.values());
        
        System.out.println("Cluster Size: " + clusters.size());
        long totaltime = (System.currentTimeMillis() - start) / 1000;
        
        System.out.println("Giving Period, Snapshot Size, Total Time (required by DBSCAN), Cluster Size, Cluster Points");
        System.out.println("WAITING: java -jar " + (int)startPeriod);
        
        FileChannel channel = null;
        try {
            // Get a file channel for the file
            channel = new RandomAccessFile(outputFile, "rw").getChannel();

            // Use the file channel to create a lock on the file.
            // This method blocks until it can retrieve the lock.
            FileLock lock = channel.lock();
            PrintWriter pw = new PrintWriter(new BufferedWriter(new FileWriter(outputFile, true)));
            
            for (CSVClusterEvent cluster : clusters) {
                pw.print((int)endPeriod + "," + nodes.keySet().size() + "," + totaltime + "," + cluster.getPoints().size() + ",");
                
                Pair<Double, Double> centroid = clusterCentroid(cluster.getPoints());
                pw.print("[" + centroid.getFirst() + ":" + centroid.getSecond() + "],\"");
                System.out.println(centroid.getFirst() + "," + centroid.getSecond());
                
                // Print the cluster mobile nodes
                List<CSVLocationUpdateEvent> points = cluster.getPoints();
                Collections.sort(points);
                String delim = "";
                for (CSVLocationUpdateEvent point : points) {
                    pw.print(delim + point.getUuid());
                    delim = ",";
                }
                pw.print("\"\n");
            }
            pw.flush();
            pw.close();
            
            // Release the lock - if it is not null!
            if( lock != null ) {
                lock.release();
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            channel.close();
        }
        System.out.println("DONE: java -jar " + (int)startPeriod);
	}
	
	public Pair<Double,Double> clusterCentroid (List<CSVLocationUpdateEvent> points) {
	    double lat = 0;
	    double lng = 0;
	    
	    for (CSVLocationUpdateEvent point : points) {
	        lat += point.getLatitude();
	        lng += point.getLongitude();
	    }
	
	   return new Pair<Double, Double>(lat/points.size(), lng/points.size());
	}
	

	/**
	 * Main Routine.
	 * 
	 * @param args a CSV filename.
	 * @throws IOException If log files cannot be created.
	 * @throws IllegalAccessException if args is null. 
	 */
	public static void main(String[] args) throws IOException {
	    // Validate args
	    if (args.length != 6) {
	        System.out.println("Usage:  java -jar CSVDBSCAN.jar csv.file     eps  minPts  time (in ms)  timewindow (in ms)  output.file");
	        System.out.println("Sample: java -jar CSVDBSCAN.jar dataset.csv  800  40      39600000      300000              out.csv");
	        System.exit(-1);
	    }
	    
		// Logging
		ConsoleAppender	consoleAppender = new ConsoleAppender(new PatternLayout("[%-4p] [%c] %m%n"));
		
		Logger.getRootLogger().setLevel(Level.INFO);
		Logger.getLogger("br.pucrio.inf.lac").addAppender(consoleAppender);
        Logger.getLogger("br.pucrio.inf.lac").setLevel(Level.INFO);

        LOGGER.info("DBSCAN CSV Clustering ...");

        String cmdLine = "";
        for (String arg : args) { 
            cmdLine += arg + " ";
        }
        LOGGER.info("CMD LINE: java -jar " + cmdLine);
        
        // CSV
        CSVDBSCAN csv = new CSVDBSCAN(args[0],                     // Input File (csv)
                                      Double.parseDouble(args[1]), // Eps
                                      Integer.parseInt(args[2]),   // minPts
                                      Double.parseDouble(args[3]), // Time
                                      Double.parseDouble(args[4]), // Time Window
                                      args[5]);                    // Output File (csv)
		csv.cluster();
	}

}
